package com.cv.infra.framework.dynamicThreadPool.core;

import com.cv.framework.common.exception.ServiceException;
import lombok.Getter;

import java.io.Serial;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Predicate;

/**
 * 自定义动态线程池队列
 * 在 LinkBlockingQueue 的基础上进行重写，以实现动态线程池的队列容量可修改
 *
 * @author Charles_XDXD
 */
public class DynamicQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    @Serial
    private static final long serialVersionUID = 1L;

    /**
     * 动态队列中的节点对象，与 LinkBlockingQueue 中的 Node 功能一致
     *
     * @param <E> 节点中存储的数据类型。
     */
    static class Node<E> {
        /**
         * 存储节点的数据项。
         * 该字段用于保存节点的实际数据，其类型为泛型类型 E。
         */
        E item;

        /**
         * 指向动态队列中的下一个节点。
         * 该字段用于将当前节点链接到队列中的下一个节点，形成链式结构。
         */
        DynamicQueue.Node<E> next;

        /**
         * 构造一个新的节点并初始化其数据项。
         *
         * @param x 要存储在节点中的数据项，其类型为泛型类型 E。
         */
        Node(E x) {
            item = x;
        }
    }


    /**
     * 队列容量。
     * 与 LinkBlockingQueue 不同的是，它可以动态地修改队列的大小，以适应动态线程池的需求。
     */
    @Getter
    private volatile int capacity;

    /**
     * 当前队列中实际存储的数据项的数量。
     */
    private final AtomicInteger count = new AtomicInteger();

    /**
     * 头节点
     */
    transient DynamicQueue.Node<E> head;

    /**
     * 尾节点
     */
    private transient DynamicQueue.Node<E> last;

    /**
     * 在 take()、poll() 等方法中，用于保护队列的并发访问。
     */
    private final ReentrantLock takeLock = new ReentrantLock();

    /**
     * 等待队列，用于等待 takes 的操作。
     */
    private final Condition notEmpty = takeLock.newCondition();

    /**
     * 在 put()、offer() 等方法中，用于保护队列的并发访问。
     */
    private final ReentrantLock putLock = new ReentrantLock();

    /**
     * 等待队列，用于等待 puts 的操作。
     */
    private final Condition notFull = putLock.newCondition();

    /**
     * 通知等待获取元素的操作。仅在 put/offer 方法中调用（这些方法通常不会锁定 takeLock）。
     */
    private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

    /**
     * 通知等待放入元素的操作。仅在 take/poll 方法中调用。
     */
    private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

    /**
     * 入列
     *
     * @param node 节点
     */
    private void enqueue(DynamicQueue.Node<E> node) {
        last = last.next = node;
    }

    /**
     * 出列
     *
     * @return 出列的数据项
     */
    private E dequeue() {
        DynamicQueue.Node<E> h = head;
        DynamicQueue.Node<E> first = h.next;
        h.next = h; // help GC
        head = first;
        E x = first.item;
        first.item = null;
        return x;
    }

    /**
     * 避免队列同时 put 和 take 操作
     */
    void fullyLock() {
        putLock.lock();
        takeLock.lock();
    }

    /**
     * 解锁，允许 put 和 take 操作
     */
    void fullyUnlock() {
        takeLock.unlock();
        putLock.unlock();
    }

    /**
     * 无参构造，默认容量为 Integer.MAX_VALUE
     */
    public DynamicQueue() {
        this(Integer.MAX_VALUE);
    }


    public DynamicQueue(int capacity) {
        if (capacity <= 0) {
            throw new ServiceException("请将队列容量保证在：1 到 " + Integer.MAX_VALUE + " 之间");
        }

        this.capacity = capacity;
        last = head = new DynamicQueue.Node<E>(null);
    }

    /**
     * 构造函数，将集合中的元素全部添加到队列中
     *
     * @param c 集合
     */
    public DynamicQueue(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            int n = 0;
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (n >= capacity)
                    throw new IllegalStateException("Queue full");
                enqueue(new DynamicQueue.Node<E>(e));
                ++n;
            }
            count.set(n);
        } finally {
            putLock.unlock();
        }
    }

    /**
     * 当前队列中实际存储的数据项的数量 - 任务数量。
     */
    public int size() {
        return count.get();
    }

    /**
     * @return 剩余容量
     */
    public int remainingCapacity() {
        return capacity - count.get();
    }

    /**
     * 将元素添加到队列中。
     *
     * @param e 要添加到队列的元素，不能为 null。
     * @throws InterruptedException 如果线程被中断。
     * @throws NullPointerException 如果尝试添加 null元素。
     */
    public void put(E e) throws InterruptedException {
        if (e == null) {
            throw new NullPointerException();
        }

        final int c;

        // 创建一个包含元素 e 的新节点。
        final DynamicQueue.Node<E> node = new DynamicQueue.Node<E>(e);

        final ReentrantLock putLock = this.putLock;
        putLock.lockInterruptibly();

        // 获取当前队列中的元素数量。
        final AtomicInteger count = this.count;

        try {
            while (count.get() >= capacity) {
                notFull.await();
            }

            // 将节点添加到队列的末尾。
            enqueue(node);

            // 更新队列中的元素数量。
            c = count.getAndIncrement();

            // 如果队列仍有空间，通知其他等待队列变满的线程。
            if (c + 1 < capacity) {
                notFull.signal();
            }
        } finally {
            putLock.unlock();
        }

        // 如果之前队列为空，通知其他等待队列变为空的线程。
        if (c == 0) {
            signalNotEmpty();
        }
    }

    /**
     * 在队列中插入指定的元素，并在指定的等待时间内阻塞，直到成功插入或超出等待时间
     *
     * @param e       要插入的元素，不能为 null
     * @param timeout 等待的时间，单位由参数 unit 指定
     * @param unit    时间单位，用于指定等待时间的粒度
     * @return 如果在指定的等待时间内成功插入元素，则返回 true；否则返回 false
     * @throws NullPointerException 如果指定的元素为 null
     * @throws InterruptedException 如果在等待过程中线程被中断
     */
    public boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException {

        if (e == null) {
            throw new NullPointerException();
        }

        // 将等待时间转换为纳秒，以便统一时间单位
        long nanos = unit.toNanos(timeout);

        final int c;

        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();

        try {
            while (count.get() >= capacity) {
                if (nanos <= 0L) {
                    return false;
                }

                nanos = notFull.awaitNanos(nanos);
            }

            enqueue(new DynamicQueue.Node<E>(e));
            c = count.getAndIncrement();

            // 如果队列仍未满，则通知其他等待插入的线程
            if (c + 1 < capacity) {
                notFull.signal();
            }
        } finally {
            putLock.unlock();
        }

        // 如果之前队列为空，则通知其他等待读取的线程
        if (c == 0) {
            signalNotEmpty();
        }

        return true;
    }

    /**
     * 将元素添加到队列中。
     * 此方法首先检查入队的元素是否为 null，如果是，则抛出 NullPointerException。
     * 如果队列已满，则返回 false，表示无法添加元素。
     * 否则，创建一个包含元素的节点，并尝试将其添加到队列中。
     * 在添加元素之前和之后，会相应地获取和释放锁，以确保线程安全。
     * 如果队列在添加元素后仍有空余空间，则通知可能正在等待队列不满的线程。
     * 如果在添加元素之前队列是空的，则通知可能正在等待队列非空的线程。
     *
     * @param e 要添加到队列中的元素，不能为 null。
     * @return 如果元素成功添加到队列中，则返回 true；如果队列已满，则返回 false。
     * @throws NullPointerException 如果指定的元素为null。
     */
    public boolean offer(E e) {
        if (e == null) {
            throw new NullPointerException();
        }

        final AtomicInteger count = this.count;

        if (count.get() == capacity) {
            return false;
        }

        final int c;
        final DynamicQueue.Node<E> node = new DynamicQueue.Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();

        try {
            if (count.get() >= capacity) {
                return false;
            }

            enqueue(node);

            c = count.getAndIncrement();

            if (c + 1 < capacity) {
                notFull.signal();
            }

        } finally {
            putLock.unlock();
        }

        if (c == 0) {
            signalNotEmpty();
        }

        return true;
    }

    /**
     * 从队列中取元素，如果队列为空，则线程会等待直到队列中有元素
     *
     * @return 队列中的一个元素
     * @throws InterruptedException 如果线程被中断
     */
    public E take() throws InterruptedException {
        final E x;
        final int c;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();

        try {
            while (count.get() == 0) {
                notEmpty.await();
            }

            x = dequeue();
            c = count.getAndDecrement();

            if (c > 1) {
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }

        if (c == capacity) {
            signalNotFull();
        }

        return x;
    }

    /**
     * 在指定时间内获取队列中的元素，如果在指定时间内队列仍然为空，则返回 null
     * 此方法会一直等待直到队列中有元素可用，或者等待时间达到指定的超时时间
     *
     * @param timeout 超时时间
     * @param unit    时间单位
     * @return 队列中的元素，如果超时或队列为空则返回 null
     * @throws InterruptedException 如果线程在等待过程中被中断
     */
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        final E x;
        final int c;
        long nanos = unit.toNanos(timeout);
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();

        try {
            while (count.get() == 0) {
                if (nanos <= 0L) {
                    return null;
                }

                nanos = notEmpty.awaitNanos(nanos);
            }

            x = dequeue();
            c = count.getAndDecrement();

            if (c > 1) {
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }

        if (c == capacity) {
            signalNotFull();
        }

        return x;
    }

    public E poll() {
        final AtomicInteger count = this.count;

        if (count.get() == 0) {
            return null;
        }

        final E x;
        final int c;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();

        try {
            if (count.get() == 0) {
                return null;
            }

            x = dequeue();
            c = count.getAndDecrement();

            if (c > 1) {
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }

        if (c == capacity) {
            signalNotFull();
        }

        return x;
    }

    public E peek() {
        final AtomicInteger count = this.count;

        if (count.get() == 0) {
            return null;
        }

        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();

        try {
            return (count.get() > 0) ? head.next.item : null;
        } finally {
            takeLock.unlock();
        }
    }

    /**
     * 从队列中移除指定节点 p 及其前驱节点 pred。
     *
     * @param p    需要移除的节点
     * @param pred 节点 p 的前驱节点
     */
    void unlink(DynamicQueue.Node<E> p, DynamicQueue.Node<E> pred) {
        // 不修改 p.next，以允许遍历 p 的迭代器保持弱一致性
        p.item = null;
        pred.next = p.next;

        if (last == p) {
            last = pred;
        }

        if (count.getAndDecrement() == capacity) {
            notFull.signal();
        }
    }

    /**
     * 从队列中移除指定元素
     * 此方法用于在队列中查找并移除与指定对象相等的元素
     *
     * @param o 要从队列中移除的对象
     * @return 如果队列中包含该元素且成功移除，则返回 true；否则返回 false
     */
    public boolean remove(Object o) {
        if (o == null) return false;
        fullyLock();
        try {
            for (DynamicQueue.Node<E> pred = head, p = pred.next;
                 p != null;
                 pred = p, p = p.next) {
                if (o.equals(p.item)) {
                    unlink(p, pred);
                    return true;
                }
            }
            return false;
        } finally {
            fullyUnlock();
        }
    }


    /**
     * 检查队列中是否包含指定元素
     * 此方法用于判断队列中是否存在与指定对象相等的元素
     *
     * @param o 要检查是否在队列中的对象
     * @return 如果队列中包含该元素，则返回 true；否则返回 false
     */
    public boolean contains(Object o) {
        if (o == null) return false;
        fullyLock();
        try {
            for (DynamicQueue.Node<E> p = head.next; p != null; p = p.next) {
                if (o.equals(p.item)) {
                    return true;
                }
            }
            return false;
        } finally {
            fullyUnlock();
        }
    }

    /**
     * 将队列中的元素转换为数组
     * 此方法创建一个包含队列中所有元素的新数组
     *
     * @return 包含队列中所有元素的数组
     */
    public Object[] toArray() {
        fullyLock();
        try {
            int size = count.get();
            Object[] a = new Object[size];
            int k = 0;

            for (DynamicQueue.Node<E> p = head.next; p != null; p = p.next) {
                a[k++] = p.item;
            }

            return a;
        } finally {
            fullyUnlock();
        }
    }

    /**
     * Returns an array containing all of the elements in this queue, in
     * proper sequence; the runtime type of the returned array is that of
     * the specified array.  If the queue fits in the specified array, it
     * is returned therein.  Otherwise, a new array is allocated with the
     * runtime type of the specified array and the size of this queue.
     *
     * <p>If this queue fits in the specified array with room to spare
     * (i.e., the array has more elements than this queue), the element in
     * the array immediately following the end of the queue is set to
     * {@code null}.
     *
     * <p>Like the {@link #toArray()} method, this method acts as bridge between
     * array-based and collection-based APIs.  Further, this method allows
     * precise control over the runtime type of the output array, and may,
     * under certain circumstances, be used to save allocation costs.
     *
     * <p>Suppose {@code x} is a queue known to contain only strings.
     * The following code can be used to dump the queue into a newly
     * allocated array of {@code String}:
     *
     * <pre> {@code String[] y = x.toArray(new String[0]);}</pre>
     * <p>
     * Note that {@code toArray(new Object[0])} is identical in function to
     * {@code toArray()}.
     *
     * @param a the array into which the elements of the queue are to
     *          be stored, if it is big enough; otherwise, a new array of the
     *          same runtime type is allocated for this purpose
     * @return an array containing all of the elements in this queue
     * @throws ArrayStoreException  if the runtime type of the specified array
     *                              is not a supertype of the runtime type of every element in
     *                              this queue
     * @throws NullPointerException if the specified array is null
     */
    @SuppressWarnings("unchecked")
    public <T> T[] toArray(T[] a) {
        fullyLock();
        try {
            int size = count.get();
            if (a.length < size)
                a = (T[]) java.lang.reflect.Array.newInstance
                        (a.getClass().getComponentType(), size);

            int k = 0;

            for (DynamicQueue.Node<E> p = head.next; p != null; p = p.next) {
                a[k++] = (T) p.item;
            }

            if (a.length > k) {
                a[k] = null;
            }
            return a;
        } finally {
            fullyUnlock();
        }
    }

    /*
     * 此方法用于清空队列中的所有元素，将队列中的元素设置为 null，并将计数器设置为 0。
     * 它还唤醒可能等待添加元素的线程，以防队列已满。
     */
    public void clear() {
        fullyLock();
        try {
            for (DynamicQueue.Node<E> p, h = head; (p = h.next) != null; h = p) {
                h.next = h;
                p.item = null;
            }

            head = last;

            if (count.getAndSet(0) >= capacity) {
                notFull.signal();
            }
        } finally {
            fullyUnlock();
        }
    }

    /**
     * @throws UnsupportedOperationException {@inheritDoc}
     * @throws ClassCastException            {@inheritDoc}
     * @throws NullPointerException          {@inheritDoc}
     * @throws IllegalArgumentException      {@inheritDoc}
     */
    public int drainTo(Collection<? super E> c) {
        return drainTo(c, Integer.MAX_VALUE);
    }

    /**
     * @throws UnsupportedOperationException {@inheritDoc}
     * @throws ClassCastException            {@inheritDoc}
     * @throws NullPointerException          {@inheritDoc}
     * @throws IllegalArgumentException      {@inheritDoc}
     */
    public int drainTo(Collection<? super E> c, int maxElements) {
        Objects.requireNonNull(c);

        if (c == this) {
            throw new IllegalArgumentException();
        }

        if (maxElements <= 0) {
            return 0;
        }

        boolean signalNotFull = false;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            int n = Math.min(maxElements, count.get());
            // count.get provides visibility to first n Nodes
            DynamicQueue.Node<E> h = head;
            int i = 0;
            try {
                while (i < n) {
                    DynamicQueue.Node<E> p = h.next;
                    c.add(p.item);
                    p.item = null;
                    h.next = h;
                    h = p;
                    ++i;
                }
                return n;
            } finally {
                // Restore invariants even if c.add() threw
                if (i > 0) {
                    // assert h.item == null;
                    head = h;
                    signalNotFull = (count.getAndAdd(-i) == capacity);
                }
            }
        } finally {
            takeLock.unlock();
            if (signalNotFull)
                signalNotFull();
        }
    }

    /**
     * Used for any element traversal that is not entirely under lock.
     * Such traversals must handle both:
     * - dequeued nodes (p.next == p)
     * - (possibly multiple) interior removed nodes (p.item == null)
     */
    DynamicQueue.Node<E> succ(DynamicQueue.Node<E> p) {
        if (p == (p = p.next))
            p = head.next;
        return p;
    }

    /**
     * Returns an iterator over the elements in this queue in proper sequence.
     * The elements will be returned in order from first (head) to last (tail).
     *
     * <p>The returned iterator is
     * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
     *
     * @return an iterator over the elements in this queue in proper sequence
     */
    public Iterator<E> iterator() {
        return new DynamicQueue.Itr();
    }

    /**
     * Weakly-consistent iterator.
     * <p>
     * Lazily updated ancestor field provides expected O(1) remove(),
     * but still O(n) in the worst case, whenever the saved ancestor
     * is concurrently deleted.
     */
    private class Itr implements Iterator<E> {
        private DynamicQueue.Node<E> next;           // Node holding nextItem
        private E nextItem;             // next item to hand out
        private DynamicQueue.Node<E> lastRet;
        private DynamicQueue.Node<E> ancestor;       // Helps unlink lastRet on remove()

        Itr() {
            fullyLock();
            try {
                if ((next = head.next) != null)
                    nextItem = next.item;
            } finally {
                fullyUnlock();
            }
        }

        public boolean hasNext() {
            return next != null;
        }

        public E next() {
            DynamicQueue.Node<E> p;
            if ((p = next) == null)
                throw new NoSuchElementException();
            lastRet = p;
            E x = nextItem;
            fullyLock();
            try {
                E e = null;
                for (p = p.next; p != null && (e = p.item) == null; )
                    p = succ(p);
                next = p;
                nextItem = e;
            } finally {
                fullyUnlock();
            }
            return x;
        }

        public void forEachRemaining(Consumer<? super E> action) {
            // A variant of forEachFrom
            Objects.requireNonNull(action);
            DynamicQueue.Node<E> p;
            if ((p = next) == null) return;
            lastRet = p;
            next = null;
            final int batchSize = 64;
            Object[] es = null;
            int n, len = 1;
            do {
                fullyLock();
                try {
                    if (es == null) {
                        p = p.next;
                        for (DynamicQueue.Node<E> q = p; q != null; q = succ(q))
                            if (q.item != null && ++len == batchSize)
                                break;
                        es = new Object[len];
                        es[0] = nextItem;
                        nextItem = null;
                        n = 1;
                    } else
                        n = 0;
                    for (; p != null && n < len; p = succ(p))
                        if ((es[n] = p.item) != null) {
                            lastRet = p;
                            n++;
                        }
                } finally {
                    fullyUnlock();
                }
                for (int i = 0; i < n; i++) {
                    @SuppressWarnings("unchecked") E e = (E) es[i];
                    action.accept(e);
                }
            } while (n > 0 && p != null);
        }

        public void remove() {
            DynamicQueue.Node<E> p = lastRet;
            if (p == null)
                throw new IllegalStateException();
            lastRet = null;
            fullyLock();
            try {
                if (p.item != null) {
                    if (ancestor == null)
                        ancestor = head;
                    ancestor = findPred(p, ancestor);
                    unlink(p, ancestor);
                }
            } finally {
                fullyUnlock();
            }
        }
    }

    /**
     * A customized variant of Spliterators.IteratorSpliterator.
     * Keep this class in sync with (very similar) LBDSpliterator.
     */
    private final class LBQSpliterator implements Spliterator<E> {
        static final int MAX_BATCH = 1 << 25;  // max batch array size;
        DynamicQueue.Node<E> current;    // current node; null until initialized
        int batch;          // batch size for splits
        boolean exhausted;  // true when no more nodes
        long est = size();  // size estimate

        LBQSpliterator() {
        }

        public long estimateSize() {
            return est;
        }

        public Spliterator<E> trySplit() {
            DynamicQueue.Node<E> h;
            if (!exhausted &&
                    ((h = current) != null || (h = head.next) != null)
                    && h.next != null) {
                int n = batch = Math.min(batch + 1, MAX_BATCH);
                Object[] a = new Object[n];
                int i = 0;
                DynamicQueue.Node<E> p = current;
                fullyLock();
                try {
                    if (p != null || (p = head.next) != null)
                        for (; p != null && i < n; p = succ(p))
                            if ((a[i] = p.item) != null)
                                i++;
                } finally {
                    fullyUnlock();
                }
                if ((current = p) == null) {
                    est = 0L;
                    exhausted = true;
                } else if ((est -= i) < 0L)
                    est = 0L;
                if (i > 0)
                    return Spliterators.spliterator
                            (a, 0, i, (Spliterator.ORDERED |
                                    Spliterator.NONNULL |
                                    Spliterator.CONCURRENT));
            }
            return null;
        }

        public boolean tryAdvance(Consumer<? super E> action) {
            Objects.requireNonNull(action);
            if (!exhausted) {
                E e = null;
                fullyLock();
                try {
                    DynamicQueue.Node<E> p;
                    if ((p = current) != null || (p = head.next) != null)
                        do {
                            e = p.item;
                            p = succ(p);
                        } while (e == null && p != null);
                    if ((current = p) == null)
                        exhausted = true;
                } finally {
                    fullyUnlock();
                }
                if (e != null) {
                    action.accept(e);
                    return true;
                }
            }
            return false;
        }

        public void forEachRemaining(Consumer<? super E> action) {
            Objects.requireNonNull(action);
            if (!exhausted) {
                exhausted = true;
                DynamicQueue.Node<E> p = current;
                current = null;
                forEachFrom(action, p);
            }
        }

        public int characteristics() {
            return (Spliterator.ORDERED |
                    Spliterator.NONNULL |
                    Spliterator.CONCURRENT);
        }
    }

    /**
     * Returns a {@link Spliterator} over the elements in this queue.
     *
     * <p>The returned spliterator is
     * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
     *
     * <p>The {@code Spliterator} reports {@link Spliterator#CONCURRENT},
     * {@link Spliterator#ORDERED}, and {@link Spliterator#NONNULL}.
     *
     * @return a {@code Spliterator} over the elements in this queue
     * @implNote The {@code Spliterator} implements {@code trySplit} to permit limited
     * parallelism.
     * @since 1.8
     */
    public Spliterator<E> spliterator() {
        return new DynamicQueue.LBQSpliterator();
    }

    /**
     * @throws NullPointerException {@inheritDoc}
     */
    public void forEach(Consumer<? super E> action) {
        Objects.requireNonNull(action);
        forEachFrom(action, null);
    }

    /**
     * Runs action on each element found during a traversal starting at p.
     * If p is null, traversal starts at head.
     */
    void forEachFrom(Consumer<? super E> action, DynamicQueue.Node<E> p) {
        // Extract batches of elements while holding the lock; then
        // run the action on the elements while not
        final int batchSize = 64;       // max number of elements per batch
        Object[] es = null;             // container for batch of elements
        int n, len = 0;
        do {
            fullyLock();
            try {
                if (es == null) {
                    if (p == null) p = head.next;
                    for (DynamicQueue.Node<E> q = p; q != null; q = succ(q))
                        if (q.item != null && ++len == batchSize)
                            break;
                    es = new Object[len];
                }
                for (n = 0; p != null && n < len; p = succ(p))
                    if ((es[n] = p.item) != null)
                        n++;
            } finally {
                fullyUnlock();
            }
            for (int i = 0; i < n; i++) {
                @SuppressWarnings("unchecked") E e = (E) es[i];
                action.accept(e);
            }
        } while (n > 0 && p != null);
    }

    /**
     * @throws NullPointerException {@inheritDoc}
     */
    public boolean removeIf(Predicate<? super E> filter) {
        Objects.requireNonNull(filter);
        return bulkRemove(filter);
    }

    /**
     * @throws NullPointerException {@inheritDoc}
     */
    public boolean removeAll(Collection<?> c) {
        Objects.requireNonNull(c);
        return bulkRemove(c::contains);
    }

    /**
     * @throws NullPointerException {@inheritDoc}
     */
    public boolean retainAll(Collection<?> c) {
        Objects.requireNonNull(c);
        return bulkRemove(e -> !c.contains(e));
    }

    /**
     * Returns the predecessor of live node p, given a node that was
     * once a live ancestor of p (or head); allows unlinking of p.
     */
    DynamicQueue.Node<E> findPred(DynamicQueue.Node<E> p, DynamicQueue.Node<E> ancestor) {
        // assert p.item != null;
        if (ancestor.item == null)
            ancestor = head;
        // Fails with NPE if precondition not satisfied
        for (DynamicQueue.Node<E> q; (q = ancestor.next) != p; )
            ancestor = q;
        return ancestor;
    }

    /**
     * Implementation of bulk remove methods.
     */
    @SuppressWarnings("unchecked")
    private boolean bulkRemove(Predicate<? super E> filter) {
        boolean removed = false;
        DynamicQueue.Node<E> p = null, ancestor = head;
        DynamicQueue.Node<E>[] nodes = null;
        int n, len = 0;
        do {
            // 1. Extract batch of up to 64 elements while holding the lock.
            fullyLock();
            try {
                if (nodes == null) {  // first batch; initialize
                    p = head.next;
                    for (DynamicQueue.Node<E> q = p; q != null; q = succ(q))
                        if (q.item != null && ++len == 64)
                            break;
                    nodes = (DynamicQueue.Node<E>[]) new DynamicQueue.Node<?>[len];
                }
                for (n = 0; p != null && n < len; p = succ(p))
                    nodes[n++] = p;
            } finally {
                fullyUnlock();
            }

            // 2. Run the filter on the elements while lock is free.
            long deathRow = 0L;       // "bitset" of size 64
            for (int i = 0; i < n; i++) {
                final E e;
                if ((e = nodes[i].item) != null && filter.test(e))
                    deathRow |= 1L << i;
            }

            // 3. Remove any filtered elements while holding the lock.
            if (deathRow != 0) {
                fullyLock();
                try {
                    for (int i = 0; i < n; i++) {
                        final DynamicQueue.Node<E> q;

                        if ((deathRow & (1L << i)) != 0L
                                && (q = nodes[i]).item != null) {
                            ancestor = findPred(q, ancestor);
                            unlink(q, ancestor);
                            removed = true;
                        }
                        nodes[i] = null; // help GC
                    }
                } finally {
                    fullyUnlock();
                }
            }
        } while (n > 0 && p != null);
        return removed;
    }

    /**
     * Saves this queue to a stream (that is, serializes it).
     *
     * @param s the stream
     * @throws java.io.IOException if an I/O error occurs
     * @serialData The capacity is emitted (int), followed by all of
     * its elements (each an {@code Object}) in the proper order,
     * followed by a null
     */
    @Serial
    private void writeObject(java.io.ObjectOutputStream s)
            throws java.io.IOException {

        fullyLock();
        try {
            // Write out any hidden stuff, plus capacity
            s.defaultWriteObject();

            // Write out all elements in the proper order.
            for (DynamicQueue.Node<E> p = head.next; p != null; p = p.next)
                s.writeObject(p.item);

            // Use trailing null as sentinel
            s.writeObject(null);
        } finally {
            fullyUnlock();
        }
    }

    /**
     * Reconstitutes this queue from a stream (that is, deserializes it).
     *
     * @param s the stream
     * @throws ClassNotFoundException if the class of a serialized object
     *                                could not be found
     * @throws java.io.IOException    if an I/O error occurs
     */
    @Serial
    private void readObject(java.io.ObjectInputStream s)
            throws java.io.IOException, ClassNotFoundException {
        // Read in capacity, and any hidden stuff
        s.defaultReadObject();

        count.set(0);
        last = head = new DynamicQueue.Node<E>(null);

        // Read in all elements and place in queue
        for (; ; ) {
            @SuppressWarnings("unchecked")
            E item = (E) s.readObject();
            if (item == null)
                break;
            add(item);
        }
    }

    /**
     * 调整当前队列容量
     *
     * @param capacity 队列容量
     */
    public void setCapacity(int capacity) {
        if (capacity <= 0) {
            throw new ServiceException("请将队列容量保证在：1 到 " + Integer.MAX_VALUE + " 之间");
        }

        this.capacity = capacity;
    }
}
